package com.gzyj.server.listener;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.gzyj.data.common.VehicleIncrement;
import com.gzyj.server.config.LogConfig;
import com.gzyj.server.factory.CommonHandler;
import com.gzyj.server.factory.jtt808.Common808HandlerFactory;
import com.gzyj.server.service.CommonService;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;

@Component
public class MonitorListener {

    @Autowired
    Common808HandlerFactory handlerFactory;

    @Value("${jtt-server.mode}")
    private String serverMode;

    @Autowired
    private CommonService commonService;

    @KafkaListener(topics = {"${kafka.queue.KafkaDownTopic}"}, groupId = "${kafka.queue.KafkaDownGroup}", containerFactory = "kafkaListenerContainerFactory")
    public void process(List<ConsumerRecord<String, String>> records) {
        if(!serverMode.equalsIgnoreCase("808")) return;

        for (ConsumerRecord<String, String> record : records) {
            String value = record.value();

            if (StringUtils.isEmpty(value)) continue;

            processSingle(value);


        }
    }


    private void processSingle(String value) {
        JSONObject command = JSON.parseObject(value);


        String ipaddress = command.getString("ipAddress");
        int commandId = command.getIntValue("cmdID");

        if (command.containsKey("sendTime")) {

            try {
                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

                Date time = sdf.parse(command.getString("sendTime"));
                long ts = (new Date()).getTime() - time.getTime();
                if (Math.abs(ts / 60000) > 30) {
                    return;
                }


            } catch (ParseException e) {
                return;
            }


        } else {
            return;
        }

        VehicleIncrement vehicleInfo = commonService.getVehInfo(ipaddress);
        if (vehicleInfo == null) {
            LogConfig.commondErrorLog.error("808 下行指令 不存在该车辆 {}", ipaddress);
            return;
        }

        CommonHandler handler = handlerFactory.getHandler(commandId);
        if (handler == null) return;
        handler.handler(vehicleInfo, value);


    }

}
